package com.example.service;

import com.example.dto.User;
import io.fury.Fury;
import io.fury.config.Language;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.queue.DistributedQueue;
import org.apache.curator.framework.recipes.queue.QueueConsumer;
import org.apache.curator.framework.recipes.queue.QueueSerializer;
import org.apache.curator.framework.state.ConnectionState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;


/***
 *
 * 测试复杂分布式的消息队列
 */
@Service
public class DistributedQueueServiceImpl implements DistributedQueueService, InitializingBean, DisposableBean {

    private DistributedQueue distributedQueue;

    public static final String DISTRIBUTEDQUEUE = "/zookeeper/queue";

    Logger logger = LoggerFactory.getLogger(getClass());

    @Resource
    private ZookeeperService zookeeperService;

    @Override
    public DistributedQueue getDistributedQueue() {
        return distributedQueue;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.distributedQueue = zookeeperService.getDistributedQueue(new QueueConsumer() {
            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {

            }

            @Override
            public void consumeMessage(Object message) throws Exception {
                if (message instanceof User) {
                    logger.warn("收到并消费消息: " + message.toString());
                }
            }
        }, new QueueSerializer() {

            private Fury fury;

            public Fury getFury() {
                if (this.fury == null) {
                    this.fury = Fury.builder().withLanguage(Language.JAVA).requireClassRegistration(true).build();
                    this.fury.register(User.class);
                }
                return this.fury;
            }

            @Override
            public byte[] serialize(Object item) {
                return getFury().serialize(item);
            }

            @Override
            public Object deserialize(byte[] bytes) {
                return getFury().deserialize(bytes);
            }
        }, DISTRIBUTEDQUEUE);
        this.distributedQueue.start();
    }

    @Override
    public void destroy() throws Exception {
        this.distributedQueue.close();
    }
}
